package org.aurora.core.server.scheduler;

import com.alibaba.fastjson.JSON;
import org.aurora.core.common.exception.AwaitHashSlotAssignException;
import org.aurora.model.request.AssignHashSlotsRequest;
import org.aurora.model.response.AssignHashSlotsResponse;
import org.aurora.remote.netty.NettyEvent;
import org.aurora.remote.netty.NettyEventSubscriber;
import org.aurora.model.type.RequestCommandType;
import org.aurora.remote.AbstractRemoteProcessor;
import org.aurora.remote.netty.NettyRemoteClient;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashSet;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class DefaultMainSchedulerConnector implements MainSchedulerConnector {
    private static final Logger log = LoggerFactory.getLogger(DefaultMainSchedulerConnector.class);

    private SchedulerConfig schedulerConfig;

    private NettyRemoteClient nettyRemoteClient;

    private MainSchedulerChangeSubscriber mainSchedulerChangeSubscriber;

    private volatile CountDownLatch awaitHashSlotAssignCountDownLatch;

    public DefaultMainSchedulerConnector(SchedulerConfig schedulerConfig, MainSchedulerChangeSubscriber mainSchedulerChangeSubscriber) {
        this.schedulerConfig = schedulerConfig;
        this.mainSchedulerChangeSubscriber = mainSchedulerChangeSubscriber;
    }

    @Override
    public void connect(String mainSchedulerIp, int mainSchedulerPort) throws InterruptedException {
        synchronized (this) {
            if (this.nettyRemoteClient != null) {
                this.nettyRemoteClient.shutdown();
            }

            // 重置状态
            this.schedulerConfig.setSchedulerHashSlots(null);
            this.awaitHashSlotAssignCountDownLatch = new CountDownLatch(1);

            this.nettyRemoteClient = new NettyRemoteClient(mainSchedulerIp, mainSchedulerPort);
            this.nettyRemoteClient.registerProcessor(RequestCommandType.SCHEDULER_ASSIGN_HASH_SLOTS.getCode(), new AssignHashSlotsProcessor());

            this.nettyRemoteClient.registerEventManager(new NettyEventSubscriber() {
                @Override
                public String getSubscriberName() {
                    return "defaultMainSchedulerConnectorSubscriber";
                }

                @Override
                public void connect(NettyEvent event) {

                }

                @Override
                public void disconnect(NettyEvent event) {
                    DefaultMainSchedulerConnector.this.mainSchedulerChangeSubscriber.change();
                }

                @Override
                public void exception(NettyEvent event) {
                    DefaultMainSchedulerConnector.this.mainSchedulerChangeSubscriber.change();
                }
            });

            this.nettyRemoteClient.start();

            this.schedulerConfig.setMainSchedulerIp(mainSchedulerIp);
            this.schedulerConfig.setMainSchedulerPort(mainSchedulerPort);
//            this.schedulerConfig.setSchedulerHashSlots(schedulerHashSlots);
        }
    }

    @Override
    public void disconnect() {
        synchronized (this) {
            if (this.nettyRemoteClient != null) {
                this.nettyRemoteClient.shutdown();
            }
        }
    }

    @Override
    public void awaitHashSlotAssign() throws InterruptedException {
        long awaitHashSlotAssignTime = 10000L;
        boolean hashSlotAssignSuccess = this.awaitHashSlotAssignCountDownLatch.await(awaitHashSlotAssignTime, TimeUnit.MILLISECONDS);

        if (!hashSlotAssignSuccess) {
            throw new AwaitHashSlotAssignException(awaitHashSlotAssignTime);
        }
    }

    class AssignHashSlotsProcessor extends AbstractRemoteProcessor<AssignHashSlotsRequest, AssignHashSlotsResponse> {

        @Override
        public AssignHashSlotsResponse doProcessor(AssignHashSlotsRequest request) {
            // todo 锁住父类这样对吗
            synchronized (DefaultMainSchedulerConnector.this) {
                log.info("receive hash slots = {}", JSON.toJSONString(request));
                DefaultMainSchedulerConnector.this.schedulerConfig.setSchedulerHashSlots(new HashSet<>(request.getSchedulerMetaData().getSlots()));
                AssignHashSlotsResponse response = new AssignHashSlotsResponse(schedulerConfig.getSchedulerIp(), schedulerConfig.getSchedulerPort());

                DefaultMainSchedulerConnector.this.awaitHashSlotAssignCountDownLatch.countDown();
                return response;
            }
        }
    }
}
